package com.xiaotu.spider;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.management.JMException;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.xiaotu.common.exception.SpiderException;
import com.xiaotu.common.exception.SpiderExceptionCode;
import com.xiaotu.common.model.SpiderLogModel;
import com.xiaotu.common.redis.RedisClientTemplate;
import com.xiaotu.common.util.GsonUtils;
import com.xiaotu.common.util.RegexUtils;
import com.xiaotu.common.util.SepratorUtil;
import com.xiaotu.common.util.SpringContextHolder;
import com.xiaotu.common.util.UUIDUtils;
import com.xiaotu.spider.downloader.AliyunProxyPool;
import com.xiaotu.spider.downloader.SpiderHttpClientDownloader;
import com.xiaotu.spider.monitor.MyStatusMXBean;
import com.xiaotu.spider.monitor.SpiderMonitorThread;

import net.sf.json.JSONObject;
import us.codecraft.webmagic.Spider;
import us.codecraft.webmagic.monitor.SpiderMonitor;
import us.codecraft.webmagic.monitor.SpiderStatusMXBean;
import us.codecraft.webmagic.scheduler.PriorityScheduler;
import us.codecraft.webmagic.scheduler.Scheduler;

/**
 * @类名 SpiderFactory
 * @日期 2017年3月6日
 * @作者 高海军
 * @功能 爬虫任务工厂类
 */
@Component
public class SpiderFactory
{
    private static final Logger LOGGER = LoggerFactory.getLogger(SpiderFactory.class);

    public static final String PAGE_PROCESSOR = "pageProcessor";// pageprocessor处理类

    public static final String PIPELINE = "pipeline";// pipeline处理类

    public static final String THREAD = "thread";// 并发线程数

    public static final String DATA_SOURCE = "dsKey";// 数据源的key

    public static final String EMPTY_DATA_SOURCE = "emptyDS";// 空数据源（初始请求地址写死在代码中，无需通过关键字来拼）

    public static final String DATA_SOURCE_RANGE = "dsRange";// 数据源的key

    public static final String SLEEP_TIME = "sleepTime";// 统一睡眠时间，单位s

    public static final String REQUEST_HEADER = "reqHeader";// 统一请求头

    public static final String SITE_REQUEST_HEADER = "siteReqHeader";// 自定义请求头

    public static final String RETRY_TIMES = "retryTimes";// 异常后重试次数

    public static final String CYCLE_RETRY_TIMES = "cycleRetry";// 循环充实次数

    public static final String RANDOM_SLEEP = "randomSleep";// 随机睡眠时间，单位s

    public static final String SET_RESPONSE = "setResp";// 是否将response放入page中

    public static final String ALIYUN_PROXY = "aProxy";// 是否使用阿里云服务器的代理

    public static final String CORRECT_FAIL_REQUEST = "correctReq";// 是否记录失败的请求，供后续批量补爬

    public static final String REFRESH_FAIL_REQUEST = "refreshCorrect";//
    // 是否刷新需要补爬的目标数据，该参数在补爬任务中使用，1为批量补爬，2为手动补爬

    public static final String FINISHED_RUNNER = "FINISHED_RUNNER";// 爬虫任务执行完成后的处理类

    public static final String URL_SCHEDULER = "urlShe";// url调度器

    public static final String CHAR_SET = "charSet";// 编码

    private static final String[] NEED_PARAS = new String[]{PAGE_PROCESSOR, PIPELINE, THREAD};

    @Autowired
    private RedisClientTemplate redisClient;

    @Autowired
    private SpiderMonitorThread monitorThread;

    public Spider runSpider(Map<String, Object> map, final SpiderLogModel spiderLog) throws
            JMException
    {
        return this.runSpider(map, spiderLog, null);
    }

    public Spider runSpider(Map<String, Object> map, final SpiderLogModel spiderLog,
            List<JSONObject> targetList)
            throws JMException
    {
        this.validate(map);

        APageProcessor processor = SpringContextHolder.getBean(
                map.get(PAGE_PROCESSOR) + "PageProcessor",
                APageProcessor.class);
        spiderLog.setJobType(processor.getSpiderType());
        setRunPara(map, processor, spiderLog);

        APipeline pipeline = SpringContextHolder
                .getBean(map.get(PIPELINE) + "Pipeline", APipeline.class);
        pipeline.setPageProcessor(processor);
        SpiderHttpClientDownloader downloader = SpringContextHolder.getBean(
                "SpiderHttpClientDownloader", SpiderHttpClientDownloader.class);

        if (targetList == null)
            targetList = this.getDataSource(spiderLog, map);

        processor.initRunParams(map);
        processor.setTargetList(targetList, spiderLog);

        Spider spider = FinsishedSpider
                .create(processor, (Runnable) map.get(FINISHED_RUNNER))
                .thread(Integer.parseInt(map.get(THREAD) + StringUtils.EMPTY));
        spider.addPipeline(pipeline);
        downloader.initDownloader(processor, spiderLog, map);
        spider.setDownloader(downloader);
        spider.setScheduler(map.containsKey(URL_SCHEDULER) ? SpringContextHolder.getBean(
                map.get(URL_SCHEDULER) + "Scheduler",
                Scheduler.class) : new PriorityScheduler());
        spider.setUUID(UUIDUtils.getStringUUID());

        SpiderMonitor spiderMonitor = new SpiderMonitor()
        {
            @Override
            protected SpiderStatusMXBean getSpiderStatusMBean(Spider spider,
                    MonitorSpiderListener monitorSpiderListener)
            {
                SpiderStatusMXBean statusMXBean = new MyStatusMXBean(spider,
                        monitorSpiderListener);
                monitorThread.init(statusMXBean, spiderLog);
                new Thread(monitorThread).start();
                return statusMXBean;
            }
        };

        spiderMonitor.register(spider);
        spider.start();
        return spider;
    }

    private List<JSONObject> getDataSource(SpiderLogModel spiderLog,
            Map<String, Object> map)
    {
        List<String> targetList = null;
        if (map.containsKey(DATA_SOURCE_RANGE))
            targetList = this.getRangeDataSource(map);
        else
            targetList = this.getAllDataSource(map);

        List<JSONObject> jsonList = new ArrayList<JSONObject>();
        for (String string : targetList)
        {
            if (StringUtils.isEmpty(string))
                continue;
            try
            {
                JSONObject jsonObj = GsonUtils.fromJson(string,
                        JSONObject.class);
                jsonList.add(0, jsonObj);
            }
            catch (Exception e)
            {
                LOGGER.error("[{}] json error:{}", spiderLog.getJobType(),
                        string, e);
            }
        }

        for (int i = 0; i < 1; i++)
        {
            //             jsonList.add(JSONObject.fromObject(
            //                                "{'name':'吴京','starid':'1131448','tvType':'1',
            // 'actor':'蔡宜达,张嘉倪,李净洋','director':'韦雄,赵箭'}"));
            //            jsonList.add(JSONObject.fromObject("{tvName:'人民的名义',tvid:'3',
            // tvType:'1',actor:'陆毅,张丰毅,吴刚,许亚军,张志坚,柯蓝,胡静,张凯丽,赵子琪,白志迪,李建义,高亚麟,丁海峰,冯雷,李光复,张晞临,徐光宇',
            // director:'李路'}"));
             //           jsonList.add(JSONObject.fromObject("{'tvName':'秘杀名单','tvid':'8','tvType':'1','actor':'靳东,李立群,周扬,刘立淇,章贺','director':'刘逢声'}"));
             //            jsonList.add(JSONObject.fromObject("{tvName:'三生三世十里桃花',tvid:'1',tvType:'1',actor:'杨幂,赵又廷,张智尧,于朦胧,黄梦莹,迪丽热巴,高伟光,张彬彬,代斯,赖艺,刘芮麟,祝绪丹,李东恒',director:'林玉芬,余翠华,任海涛'}"));
            // jsonList.add(JSONObject.fromObject("{starid:'1320',name:'杨千嬅'}"));
            //            jsonList.add(JSONObject.fromObject("{'tvName':'鬼吹灯之黄皮子坟',
            // 'tvid':'1120497','tvType':'3',
            // 'actor':'阮经天,徐璐," +
            //                    "郝好,刘潮,李玉洁','director':'管虎,费振翔'}"));
            //            jsonList.add(null);
        }

        return jsonList;
    }

    private void validate(Map<String, Object> map)
    {
        if (map == null || map.isEmpty())
            throw new SpiderException(
                    SpiderExceptionCode.PageProcessorParameterErrorException,
                    "input param is error");
        for (String para : NEED_PARAS)
            if (map.get(para) == null)
                throw new SpiderException(
                        SpiderExceptionCode.PageProcessorParameterErrorException,
                        "input param is error");
    }

    private void setRunPara(Map<String, Object> map, APageProcessor processor,
            SpiderLogModel spiderLog)
    {
        if (map.containsKey(SpiderFactory.REFRESH_FAIL_REQUEST))
            spiderLog.setJobNature(
                    Integer.parseInt(map.get(SpiderFactory.REFRESH_FAIL_REQUEST)
                            + StringUtils.EMPTY));

        String sleepTime = map.get(SLEEP_TIME) == null ? null
                : map.get(SLEEP_TIME) + StringUtils.EMPTY;
        if (StringUtils.isNotEmpty(sleepTime)
                && RegexUtils.regexMatch(RegexUtils.REGEX_INTEGER, sleepTime))
            processor.setSleepTime(Integer.parseInt(sleepTime));
        String retryTimes = map.get(RETRY_TIMES) == null ? null
                : map.get(RETRY_TIMES) + StringUtils.EMPTY;
        if (StringUtils.isNotEmpty(retryTimes)
                && RegexUtils.regexMatch(RegexUtils.REGEX_INTEGER, retryTimes))
            processor.setRetryTimes(Integer.parseInt(retryTimes));

        retryTimes = map.get(CYCLE_RETRY_TIMES) == null ? null
                : map.get(CYCLE_RETRY_TIMES) + StringUtils.EMPTY;
        if (StringUtils.isNotEmpty(retryTimes)
                && RegexUtils.regexMatch(RegexUtils.REGEX_INTEGER, retryTimes))
            processor.setCycleRetryTimes(Integer.parseInt(retryTimes));

        if (map.containsKey(SpiderFactory.REQUEST_HEADER))
            processor.setRequestHeader();
        if (map.containsKey(SpiderFactory.SITE_REQUEST_HEADER))
            processor.setSiteRequestHeader();

        if (map.containsKey(SpiderFactory.ALIYUN_PROXY))
            processor.getSite().setHttpProxyPool(new AliyunProxyPool());

        if (map.containsKey(SpiderFactory.CHAR_SET))
            processor.setCharSet(map.get(SpiderFactory.CHAR_SET) + StringUtils.EMPTY);
    }

    private List<String> getAllDataSource(Map<String, Object> map)
    {
        String dataSource = map.get(DATA_SOURCE) + StringUtils.EMPTY;
        if (EMPTY_DATA_SOURCE.equals(dataSource))
            return Arrays.asList(new String[]{StringUtils.EMPTY});

        Map<String, String> targetMap = redisClient.hgetAll(dataSource);

        if (targetMap == null || targetMap.isEmpty())
            throw new SpiderException(
                    SpiderExceptionCode.PageProcessorParameterErrorException,
                    "data source is empty:" + map.get(DATA_SOURCE));

        return new ArrayList<String>(targetMap.values());
    }

    private List<String> getRangeDataSource(Map<String, Object> map)
    {
        Set<String> keySet = redisClient
                .hkeys(map.get(DATA_SOURCE) + StringUtils.EMPTY);
        if (keySet == null || keySet.isEmpty())
            throw new SpiderException(
                    SpiderExceptionCode.PageProcessorParameterErrorException,
                    "data source is empty:" + map.get(DATA_SOURCE));
        String range = map.get(DATA_SOURCE_RANGE) + StringUtils.EMPTY;
        if (StringUtils.isEmpty(range)
                || !range.contains(SepratorUtil.SEP_COLON_EN))
            throw new SpiderException(
                    SpiderExceptionCode.PageProcessorParameterErrorException,
                    "data source range is error:" + range);

        String[] indexs = range.split(SepratorUtil.SEP_COLON_EN);
        if (indexs.length != 2
                || !RegexUtils.regexMatch(RegexUtils.REGEX_INTEGER, indexs[0]))
            throw new SpiderException(
                    SpiderExceptionCode.PageProcessorParameterErrorException,
                    "data source range is error:" + range);

        int start = Integer.parseInt(indexs[0]);
        int end = Integer.parseInt(indexs[1]);
        end = end >= keySet.size() ? end = keySet.size() - 1 : end;
        end = end == -1 ? keySet.size() - 1 : end;
        if (start < 0 || start >= end)
            throw new SpiderException(SpiderExceptionCode.PageProcessorParameterErrorException,
                    "data source range is error:" + range);

        String[] keyArr = new String[keySet.size()];
        keySet.toArray(keyArr);
        String[] resArr = new String[end - start + 1];
        System.arraycopy(keyArr, start, resArr, 0, end - start + 1);
        return redisClient.hmget(map.get(DATA_SOURCE) + StringUtils.EMPTY, resArr);
    }
}
